package net.bwie.realtime.jtp.dws.trade.job;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;

/**
 * @Author: FuHe
 * @Date: 2025/6/3
 */
public class JtpOrderSkuAggregateDwsJob {
    public static void main(String[] args) {
        // 1 表执行环境
        TableEnvironment tabEnv = getTableEnv();
        // 2 输入表
        createInputTable(tabEnv);
        // 3 数据处理 select
        Table reportTable = handle(tabEnv);
        // 4 输出表 输出到doris
        createOutputTable(tabEnv);
        // 5 保存到doris中
        saveToDoris(tabEnv, reportTable);

    }

    private static void saveToDoris(TableEnvironment tabEnv, Table reportTable) {
        tabEnv.createTemporaryView( "report_table", reportTable);
        tabEnv.executeSql("INSERT INTO dws_trade_sku_order_report_doris_sink\n" +
                "SELECT SUBSTRING(CAST(start_time AS STRING), 0, 19) AS start_time,\n" +
                "       SUBSTRING(CAST(end_time AS STRING), 0, 19)   AS end_time,\n" +
                "       SUBSTRING(CAST(start_time AS STRING), 0, 10) AS cur_date,\n" +
                "       sku_id,\n" +
                "       sku_name,\n" +
                "       spu_id,\n" +
                "       spu_name,\n" +
                "       tm_id,\n" +
                "       tm_name,\n" +
                "       category3_id,\n" +
                "       category3_name,\n" +
                "       category2_id,\n" +
                "       category2_name,\n" +
                "       category1_id,\n" +
                "       category1_name,\n" +
                "       order_count,\n" +
                "       order_amount\n" +
                "FROM report_table");
    }

    private static void createOutputTable(TableEnvironment tabEnv) {
        tabEnv.executeSql("CREATE TABLE dws_trade_sku_order_report_doris_sink\n" +
                "(\n" +
                "    `start_time`     STRING,\n" +
                "    `end_time`       STRING,\n" +
                "    `cur_date`       STRING,\n" +
                "    `sku_id`         STRING,\n" +
                "    `sku_name`       STRING,\n" +
                "    `spu_id`         STRING,\n" +
                "    `spu_name`       STRING,\n" +
                "    `tm_id`          STRING,\n" +
                "    `tm_name`        STRING,\n" +
                "    `category3_id`   STRING,\n" +
                "    `category3_name` STRING,\n" +
                "    `category2_id`   STRING,\n" +
                "    `category2_name` STRING,\n" +
                "    `category1_id`   STRING,\n" +
                "    `category1_name` STRING,\n" +
                "    `order_count`    BIGINT,\n" +
                "    `order_amount`   DECIMAL(16, 2)\n" +
                ") WITH (\n" +
                "    'connector' = 'doris',\n" +
                "    'fenodes' = 'node102:8030',\n" +
                "    'table.identifier' = 'jtp_mall_report.dws_trade_sku_order_report',\n" +
                "    'username' = 'root',\n" +
                "    'password' = '123456',\n" +
                "    'sink.batch.interval' = '10s',\n" +
                "    'sink.max-retries' = '3',\n" +
                "    'sink.batch.size' = '1000'\n" +
                ")");
    }

    private static Table handle(TableEnvironment tabEnv) {
        Table orderJoinTable = tabEnv.sqlQuery("SELECT t1.id,\n" +
                "       t1.order_id,\n" +
                "       t1.user_id,\n" +
                "       t1.sku_id,\n" +
                "       t1.sku_name,\n" +
                "       t1.province_id,\n" +
                "       t1.create_time,\n" +
                "       t1.source_type_name,\n" +
                "       t1.sku_num,\n" +
                "       t1.split_original_amount,\n" +
                "       t1.split_total_amount,\n" +
                "       t2.spu_id,\n" +
                "       t2.category3_id,\n" +
                "       t2.tm_id\n" +
                "FROM dwd_order_detail_kafka_source AS t1\n" +
                "         LEFT JOIN dim_sku_info_hbase_source FOR SYSTEM_TIME AS OF t1.proc_time AS t2\n" +
                "ON t1.sku_id = t2.row_key");
        tabEnv.createTemporaryView("order_join_table", orderJoinTable);
        Table resultTable = tabEnv.sqlQuery("SELECT\n" +
                "    DATE_FORMAT(window_start, 'yyyy-MM-dd HH:mm:ss') AS  start_time,\n" +
                "    DATE_FORMAT(window_end, 'yyyy-MM-dd HH:mm:ss') AS  end_time,\n" +
                "    sku_id,sku_name,spu_id,category3_id,tm_id,\n" +
                "    count(distinct id) AS order_count,\n" +
                "    sum(cast(split_total_amount AS DECIMAL(10,2))) AS order_amount,\n" +
                "    PROCTIME() AS proc_time\n" +
                "FROM TABLE ( TUMBLE(DATA => TABLE order_join_table,\n" +
                "    TIMECOL => DESCRIPTOR(create_time),\n" +
                "    SIZE => INTERVAL '1' MINUTES) )\n" +
                "WHERE user_id IS NOT NULL AND source_type_name IS NOT NULL\n" +
                "GROUP BY window_start,  window_end, sku_id,sku_name,spu_id,category3_id,tm_id");
        tabEnv.createTemporaryView("result_table", resultTable);
        Table reportTable = tabEnv.sqlQuery("SELECT t1.start_time,\n" +
                "       t1.end_time,\n" +
                "       t1.sku_id,\n" +
                "       t1.sku_name,\n" +
                "       t1.spu_id,\n" +
                "       t2.spu_name,\n" +
                "       t1.tm_id,\n" +
                "       t3.tm_name,\n" +
                "       t1.category3_id,\n" +
                "       t4.name                 AS category3_name,\n" +
                "       t4.category2_id,\n" +
                "       t5.name                 AS category2_name,\n" +
                "       t5.category1_id,\n" +
                "       t6.name                 AS category1_name,\n" +
                "       t1.order_count,\n" +
                "       t1.order_amount,\n" +
                "       UNIX_TIMESTAMP() * 1000 AS ts\n" +
                "FROM result_table t1\n" +
                "         LEFT JOIN dim_spu_info_hbase_source FOR SYSTEM_TIME AS OF t1.proc_time AS t2\n" +
                "ON t1.spu_id = t2.row_key\n" +
                "    LEFT JOIN dim_base_trademark_hbase_source FOR SYSTEM_TIME AS OF t1.proc_time AS t3 ON t1.tm_id = t3.row_key\n" +
                "    LEFT JOIN dim_base_category3_hbase_source FOR SYSTEM_TIME AS OF t1.proc_time AS t4 ON t1.category3_id = t4.row_key\n" +
                "    LEFT JOIN dim_base_category2_hbase_source FOR SYSTEM_TIME AS OF t1.proc_time AS t5 ON t4.category2_id = t5.row_key\n" +
                "    LEFT JOIN dim_base_category1_hbase_source FOR SYSTEM_TIME AS OF t1.proc_time AS t6 ON t5.category1_id = t6.row_key");
        return reportTable;
    }

    private static void createInputTable(TableEnvironment tabEnv) {
        tabEnv.executeSql("CREATE TABLE dwd_order_detail_kafka_source\n" +
                "(\n" +
                "    id                    STRING,\n" +
                "    order_id              STRING,\n" +
                "    user_id               STRING,\n" +
                "    order_status          STRING,\n" +
                "    sku_id                STRING,\n" +
                "    sku_name              STRING,\n" +
                "    province_id           STRING,\n" +
                "    activity_id           STRING,\n" +
                "    activity_rule_id      STRING,\n" +
                "    coupon_id             STRING,\n" +
                "    date_id               STRING,\n" +
                "    create_time           TIMESTAMP(3),\n" +
                "    operate_date_id       STRING,\n" +
                "    operate_time          STRING,\n" +
                "    source_id             STRING,\n" +
                "    source_type           STRING,\n" +
                "    source_type_name      STRING,\n" +
                "    sku_num               STRING,\n" +
                "    split_original_amount STRING,\n" +
                "    split_activity_amount STRING,\n" +
                "    split_coupon_amount   STRING,\n" +
                "    split_total_amount    STRING,\n" +
                "    row_op_ts TIMESTAMP_LTZ(3),\n" +
                "    proc_time AS PROCTIME(),\n" +
                "    WATERMARK FOR create_time AS create_time - INTERVAL '0' MINUTE\n" +
                ")WITH(\n" +
                "    'connector' = 'kafka',\n" +
                "    'topic' = 'dwd-order-detail',\n" +
                "    'properties.bootstrap.servers' = 'node101:9092,node102:9092,node103:9092',\n" +
                "    'properties.group.id' = 'gid-dws-trade-sku-order',\n" +
                "    'scan.startup.mode' = 'earliest-offset',\n" +
                "    'format' = 'json',\n" +
                "    'json.fail-on-missing-field' = 'false',\n" +
                "    'json.ignore-parse-errors' = 'true'\n" +
                ")");
        tabEnv.executeSql("CREATE TABLE dim_sku_info_hbase_source\n" +
                "(\n" +
                "    row_key         STRING,\n" +
                "    info ROW <spu_id STRING,\n" +
                "    price           STRING,\n" +
                "    sku_name        STRING,\n" +
                "    sku_desc        STRING,\n" +
                "    weight          STRING,\n" +
                "    tm_id           STRING,\n" +
                "    category3_id    STRING,\n" +
                "    sku_default_img STRING,\n" +
                "    is_sale         STRING,\n" +
                "    create_time     STRING>,\n" +
                "    PRIMARY KEY (row_key) NOT ENFORCED\n" +
                ") WITH (\n" +
                "    'connector' = 'hbase-2.2',\n" +
                "    'table-name' = 'dim_sku_info',\n" +
                "    'zookeeper.quorum' = 'node101:2181,node102:2181,node103:2181',\n" +
                "    'lookup.async' = 'true',\n" +
                "    'lookup.cache.max-rows' = '10',\n" +
                "    'lookup.cache.ttl' = '1 hour'\n" +
                ")");
        tabEnv.executeSql("CREATE TABLE dim_spu_info_hbase_source\n" +
                "(\n" +
                "    row_key STRING,\n" +
                "    info ROW<id STRING, spu_name STRING, description STRING, category3_id STRING, tm_id STRING>,\n" +
                "    PRIMARY KEY (row_key) NOT ENFORCED\n" +
                ") WITH (\n" +
                "    'connector' = 'hbase-2.2',\n" +
                "    'table-name' = 'dim_spu_info',\n" +
                "    'zookeeper.quorum' = 'node101:2181,node102:2181,node103:2181',\n" +
                "    'lookup.async' = 'true',\n" +
                "    'lookup.cache.max-rows' = '10',\n" +
                "    'lookup.cache.ttl' = '1 hour'\n" +
                ")");
        tabEnv.executeSql("CREATE TABLE dim_base_trademark_hbase_source\n" +
                "(\n" +
                "    row_key  STRING,\n" +
                "    info ROW <id STRING, tm_name  STRING, logo_url STRING>,\n" +
                "    PRIMARY KEY (row_key) NOT ENFORCED\n" +
                ") WITH (\n" +
                "    'connector' = 'hbase-2.2',\n" +
                "    'table-name' = 'dim_base_trademark',\n" +
                "    'zookeeper.quorum' = 'node101:2181,node102:2181,node103:2181',\n" +
                "    'lookup.async' = 'true',\n" +
                "    'lookup.cache.max-rows' = '10',\n" +
                "    'lookup.cache.ttl' = '1 hour'\n" +
                ")");
        tabEnv.executeSql("CREATE TABLE dim_base_category3_hbase_source\n" +
                "(\n" +
                "    row_key      STRING,\n" +
                "    info ROW <id STRING, name STRING,\n" +
                "    category2_id STRING>,\n" +
                "    PRIMARY KEY (row_key) NOT ENFORCED\n" +
                ") WITH (\n" +
                "    'connector' = 'hbase-2.2',\n" +
                "    'table-name' = 'dim_base_category3',\n" +
                "    'zookeeper.quorum' = 'node101:2181,node102:2181,node103:2181',\n" +
                "    'lookup.async' = 'true',\n" +
                "    'lookup.cache.max-rows' = '10',\n" +
                "    'lookup.cache.ttl' = '1 hour'\n" +
                ")");
        tabEnv.executeSql("CREATE TABLE dim_base_category2_hbase_source\n" +
                "(\n" +
                "    row_key      STRING,\n" +
                "    info ROW <id STRING, name STRING,\n" +
                "    category1_id STRING>,\n" +
                "    PRIMARY KEY (row_key) NOT ENFORCED\n" +
                ") WITH (\n" +
                "    'connector' = 'hbase-2.2',\n" +
                "    'table-name' = 'dim_base_category2',\n" +
                "    'zookeeper.quorum' = 'node101:2181,node102:2181,node103:2181',\n" +
                "    'lookup.async' = 'true',\n" +
                "    'lookup.cache.max-rows' = '10',\n" +
                "    'lookup.cache.ttl' = '1 hour'\n" +
                ")");
        tabEnv.executeSql("CREATE TABLE dim_base_category1_hbase_source\n" +
                "(\n" +
                "    row_key STRING,\n" +
                "    info ROW <id STRING, name STRING>,\n" +
                "    PRIMARY KEY (row_key) NOT ENFORCED\n" +
                ") WITH (\n" +
                "    'connector' = 'hbase-2.2',\n" +
                "    'table-name' = 'dim_base_category1',\n" +
                "    'zookeeper.quorum' = 'node101:2181,node102:2181,node103:2181',\n" +
                "    'lookup.async' = 'true',\n" +
                "    'lookup.cache.max-rows' = '10',\n" +
                "    'lookup.cache.ttl' = '1 hour'\n" +
                ")");
    }

    private static TableEnvironment getTableEnv() {
        // 环境属性设置
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .useBlinkPlanner()
                .inStreamingMode()
                .build();
        // 创建表执行环境
        TableEnvironment tabEnv = TableEnvironment.create(settings);
        // 配置属性设置
        Configuration configuration = tabEnv.getConfig().getConfiguration();
        // 时区设置
        configuration.setString("table.local-time-zone", "Asia/Shanghai");
        // 并发设置
        configuration.setString("table.exec.resource.default-parallelism", "1");
        // 状态保存时间
        configuration.setString("table.exec.state.ttl", "5s");
        return tabEnv;
    }
}
